package com.atguigu;

import com.atguigu.realtime.bean.TableProcess;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/2/5 8:41
 */
public class FlinkSqlCDC_1 {
    public static void main(String[] args) {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        final StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        tenv
            .executeSql("CREATE TABLE `table_process`( " +
                            "   `source_table`  string, " +
                            "   `operate_type`  string, " +
                            "   `sink_type`  string, " +
                            "   `sink_table`  string, " +
                            "   `sink_columns` string, " +
                            "   `sink_pk`  string, " +
                            "   `sink_extend`  string, " +
                            "   PRIMARY KEY (`source_table`,`operate_type`)  NOT ENFORCED" +
                            ")with(" +
                            "   'connector' = 'mysql-cdc', " +
                            "   'hostname' = 'hadoop162', " +
                            "   'port' = '3306', " +
                            "   'username' = 'root', " +
                            "   'password' = 'aaaaaa', " +
                            "   'database-name' = 'gmall2021_realtime', " +
                            "   'table-name' = 'table_process'," +
                            "   'debezium.snapshot.mode' = 'initial' " +  // 读取mysql的全量,增量以及更新数据
                            ")");
        final Table table = tenv.sqlQuery("select " +
                                              "  source_table sourceTable, " +
                                              "  sink_type sinkType, " +
                                              "  operate_type operateType, " +
                                              "  sink_table sinkTable, " +
                                              "  sink_columns sinkColumns, " +
                                              "  sink_pk sinkPk, " +
                                              "  sink_extend sinkExtend " +
                                              "from table_process ");
        final SingleOutputStreamOperator<TableProcess> tableProcessStream = tenv
            .toRetractStream(table, TableProcess.class)
            .filter(t -> t.f0)
            .map(t -> t.f1);
        tableProcessStream.print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
